diff --git a/test/source-sink-errors/mzcompose.py b/test/source-sink-errors/mzcompose.py
index ac2b7c3e92..5a9bbeb7ee 100644
--- a/test/source-sink-errors/mzcompose.py
+++ b/test/source-sink-errors/mzcompose.py
@@ -13,6 +13,7 @@ Disruption and then checking the mz_internal.mz_*_statuses tables
 """

 import random
+import time
 from collections.abc import Callable
 from dataclasses import dataclass
 from textwrap import dedent
@@ -561,3 +562,52 @@ def unsupported_pg_table(c: Composition) -> None:
                  $ postgres-execute connection=postgres://postgres:postgres@postgres
                  INSERT INTO source1 VALUES (3, '[2:3]={2,2}')
                  """))
+
+
+def workflow_test_pk_violation_warning(c: Composition) -> None:
+    seed = random.randint(0, 256**4)
+    c.up("redpanda", "materialized", Service("testdrive", idle=True))
+
+    with c.override(
+        Testdrive(
+            no_reset=True,
+            seed=seed,
+            entrypoint_extra=["--initial-backoff=1s", "--backoff-factor=0"],
+        )
+    ):
+        c.testdrive(dedent("""
+                > CREATE CONNECTION kafka_conn
+                  TO KAFKA (BROKER '${testdrive.kafka-addr}', SECURITY PROTOCOL PLAINTEXT);
+
+                > CREATE CONNECTION IF NOT EXISTS csr_conn TO CONFLUENT SCHEMA REGISTRY (
+                  URL '${testdrive.schema-registry-url}'
+                  );
+
+                > CREATE TABLE t (a INT, b INT);
+
+                > CREATE SINK pk_test_sink FROM t
+                  INTO KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-pk-test-${testdrive.seed}')
+                  KEY (a) NOT ENFORCED
+                  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn
+                  ENVELOPE UPSERT;
+
+                > SELECT status FROM mz_internal.mz_sink_statuses WHERE name = 'pk_test_sink'
+                running
+        """))
+
+        # PkViolationWarner rate-limits to once per 10s from construction.
+        time.sleep(12)
+
+        c.testdrive(dedent("""
+                > INSERT INTO t VALUES (1, 10), (1, 20);
+
+                $ set-sql-timeout duration=60s
+                $ kafka-verify-topic sink=materialize.public.pk_test_sink await-value-schema=true
+        """))
+
+        time.sleep(5)
+
+        logs = c.invoke("logs", "materialized", capture=True)
+        assert "primary key error" in logs.stdout, (
+            "PkViolationWarner::flush() not called after batch"
+        )